package com.bw.test2;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang.Validate;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class UserLoginDWS {

    public static void main(String[] args) throws Exception {

        //独立    7日回流
        /*
        *   独立访客我我们通过   状态编程   来进行区分（  你来过我就给你记录下来  后面在来 有依据了）
        *   7日回流  也是通过状态来进行 区分（状态中的时候  和你当前时间差  >=8的时候）
        * */
        //zs   dl    hl
        //zs  dl=1    hl=0
        //zs 过滤掉
        //zs 过滤掉
        //第二天来了
        //zs   dl=1  hl=0
        //zs 过滤掉
        //第十天来了
        //zs   dl=1  hl=1
        //zs
        //zs
        //zs
        //zs
        //zs
        //第十一天
        //zs   dl=1   dl=0

        //1.创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.设置并行度
        //3.读取页面数据
        DataStreamSource<String> pageDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_page_log", "dwd_traffic_page_log_groups"));
        //4.过滤数据
        SingleOutputStreamOperator<JSONObject> flatMapDS = pageDS.flatMap(new RichFlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {

                JSONObject jsonObject = JSON.parseObject(value);

                String uid = jsonObject.getJSONObject("common").getString("uid");
                String last_page_id = jsonObject.getJSONObject("page").getString("last_page_id");
                if (uid != null && (last_page_id == null || last_page_id.equals("login"))) {
                    out.collect(jsonObject);
                }
            }
        });
        //5.设置水位线
        SingleOutputStreamOperator<JSONObject> pageLoginWaterMarkDS = flatMapDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts");
                    }
                }));
        //5. 统计   独立   统计 回流用了
        KeyedStream<JSONObject, String> keyedStream = pageLoginWaterMarkDS.keyBy(a -> a.getJSONObject("common").getString("uid"));
        //6.调用处理函数       开始时间  结束时间   hl   dl  系统时间
        SingleOutputStreamOperator<UserLoginBean> userLoginBeanSingleOutputStreamOperator = keyedStream.
                flatMap(new RichFlatMapFunction<JSONObject, UserLoginBean>() {
            ValueState<String> valueState;
            @Override
            public void open(Configuration parameters) throws Exception {
                valueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("valueState", String.class));
            }

            @Override
            public void flatMap(JSONObject value, Collector<UserLoginBean> out) throws Exception {
                //1.取出状态的值   进行判断你是否是独立用户   回流用户
                String stateDate = valueState.value();
                Long ts = value.getLong("ts");
                String currentDate = DateFormatUtil.toDate(ts);

                Long back_ct = 0l;
                Long uu_ct = 0l;
                if (stateDate == null) {
                    uu_ct = 1l;
                    valueState.update(currentDate);
                } else {
                    if (!stateDate.equals(currentDate)) {
                        uu_ct = 1l;
                        valueState.update(currentDate);
                        //当前的时间戳-状态时间戳     /天   >=8
                        Long curTS = DateFormatUtil.toTs(currentDate);
                        Long stateTS = DateFormatUtil.toTs(stateDate);
                        if ((curTS - stateTS) / (1000 * 60 * 60 * 24) >= 8) {
                            back_ct = 1l;
                        }
                    }

                }

                if (uu_ct != 0l) {
                    out.collect(new UserLoginBean("", "", back_ct, uu_ct, 0l));
                }

            }
        });
        //7.分组聚合  求0-10秒内容的  独立用户数  回流用户数
        SingleOutputStreamOperator<UserLoginBean> operator = userLoginBeanSingleOutputStreamOperator.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<UserLoginBean>() {
                    @Override
                    public UserLoginBean reduce(UserLoginBean value1, UserLoginBean value2) throws Exception {
                        value1.setBack_ct(value1.getBack_ct() + value2.getBack_ct());
                        value1.setUu_ct(value1.getUu_ct() + value2.getUu_ct());
                        return value1;
                    }
                }, new AllWindowFunction<UserLoginBean, UserLoginBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<UserLoginBean> values, Collector<UserLoginBean> out) throws Exception {
                        UserLoginBean userLogin = values.iterator().next();
                        userLogin.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        userLogin.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        userLogin.setTs(System.currentTimeMillis());
                        out.collect(userLogin);
                    }
                });
        operator.addSink(MyClickHouseUtil.getSinkFunction("insert  into   dws_user_user_login_window values(?,?,?,?,?)"));
        env.execute();

    }



}
